705d64e06186ee244d46df7ab48e0ba44ffcc442,bridge/src/main/java/io/rhiot/kafka/bridge/RawMessageConverter.java,RawMessageConverter,toKafkaRecord,#String#Message#,43

Before Change


		byte[] buffer = new byte[RawMessageConverter.BUFFER_SIZE];
		
		// get topic and body from AMQP message
		String topic = message.getAddress();
		if (topic == null) {
			topic = kafkaTopic;
		}
		

After Change


		byte[] buffer = new byte[RawMessageConverter.BUFFER_SIZE];
		
		// get topic and body from AMQP message
		String topic = (message.getAddress() == null) ?
				kafkaTopic :
				message.getAddress().replace('/', '.');
		
		int encoded = message.encode(buffer, 0, RawMessageConverter.BUFFER_SIZE);
		value = Arrays.copyOfRange(buffer, 0, encoded);